package cn.iocoder.springboot.lab03.kafkademo.consumer;

import cn.iocoder.springboot.lab03.kafkademo.message.Demo02Message;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
public class Demo02Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = Demo02Message.TOPIC,
            groupId = "demo02-consumer-group-" + Demo02Message.TOPIC)
    public void onMessage(List<ConsumerRecord<String, String>> messages, Consumer consumer) {
        logger.info("[onMessage][线程编号:{} 消息数量：{}]", Thread.currentThread().getId(), messages.size());

        logger.info("consumerBatch =====> messageSize: {}", messages.size());
        Map<Integer, List<ConsumerRecord<String, String>>> map = new HashMap<>();

        for (ConsumerRecord<String, String> message : messages) {
            List<ConsumerRecord<String, String>> consumerRecords = null;
            if (map.containsKey(message.partition())) {
                consumerRecords = map.get(message.partition());
                consumerRecords.add(message);
            }
            else {
                consumerRecords = new ArrayList<>();
                consumerRecords.add(message);
                map.put(message.partition(), consumerRecords);
            }
        }

        Set<Map.Entry<Integer, List<ConsumerRecord<String, String>>>> entries = map.entrySet();
        for (Map.Entry<Integer, List<ConsumerRecord<String, String>>> entry : entries) {
            List<ConsumerRecord<String, String>> value = entry.getValue();
            for (ConsumerRecord<String, String> stringStringConsumerRecord : value) {
                logger.info(stringStringConsumerRecord.toString());
            }

            long poff = value.get(value.size() - 1).offset() + 1;
            OffsetAndMetadata pom = new OffsetAndMetadata(poff);
            Map<TopicPartition, OffsetAndMetadata> mapP = new HashMap<>();

            TopicPartition topicPartition = new TopicPartition(Demo02Message.TOPIC, entry.getKey());

            mapP.put(topicPartition, pom);

            consumer.commitSync(mapP);
            logger.info("分区{} 已提交，offset={}", entry.getKey(), poff);
        }

    }

}
